package com.ule.merchant.order.mon.service.utils;

import com.ule.tools.client.kafka.core.config.KafkaClientsConfig;
import com.ule.tools.client.kafka.producer.Producer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.HashMap;

public class KafkaProducerUtil {

	private static Log log = LogFactory.getLog(KafkaProducerUtil.class);

	/**
	 * kafka 生产者发送Map类型消息
	 * @param producerId 生产者id
	 * @param key
	 * @param mapParameters 发送的参数
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public static void sendKafkaProducerByMap(String producerId, String key,HashMap<String, Object> mapParameters) {
		try {
			// 获取默认路径下的kafka-clients-cfg.xml
			KafkaClientsConfig kafkaConfig = KafkaClientsConfig.DEFAULT_INSTANCE;
			// 参数为配置文件中Producer的id内容
			Producer<HashMap> producer= (Producer<HashMap>) kafkaConfig.getOrCreateProducer(producerId);
			producer.send(key, mapParameters);
		} catch (Exception e) {
			log.error("异步发送kafaka消息失败,producerID:"+producerId);
			log.error(e);
		}
	}
	
	/**
	 * kafka 生产者发送Map-String类型消息
	 * @param producerId 生产者id
	 * @param key
	 * @param mapParameters 发送的参数
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public static void sendKafkaProducerByMapStr(String producerId, String key, HashMap<String, String> mapParameters) {
		try {
			// 获取默认路径下的kafka-clients-cfg.xml
			KafkaClientsConfig kafkaConfig = KafkaClientsConfig.DEFAULT_INSTANCE;
			// 参数为配置文件中Producer的id内容
			Producer<HashMap> producer= (Producer<HashMap>) kafkaConfig.getOrCreateProducer(producerId);
			producer.send(key, mapParameters);
		} catch (Exception e) {
			log.error("异步发送kafaka消息失败,producerID:"+producerId);
			log.error(e.getMessage());
		}
	}
	
	/**
	 * kafka 生产者发送json消息
	 * @param producerId 生产者id
	 * @param key
	 * @param jsonStr 发送的参数
	 */
	@SuppressWarnings("unchecked")
	public static void sendKafkaProducerByJson(String producerId, String key, String jsonStr) {
		try {
			// 获取默认路径下的kafka-clients-cfg.xml
			KafkaClientsConfig kafkaConfig = KafkaClientsConfig.DEFAULT_INSTANCE;
			// 参数为配置文件中Producer的id内容
			Producer<String> producer= (Producer<String>) kafkaConfig.getOrCreateProducer(producerId);
			producer.send(key, jsonStr);
			log.info("异步发送kafaka消息成功,producerID:" + producerId);
		} catch (Exception e) {
			log.error("异步发送kafaka消息失败,producerID:" + producerId + "," + e.getMessage());
		}
	}

}
